package com.eshipenet.shipnet.equipmentextbuilder

import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.util.Collector

class EquipmentWithModelIdActionGenerator
    extends CoProcessFunction[EquipmentAction, EquipmentModelLinkAction, EquipmentWithModelIdAction] {
  private lazy val equipmentDesc =
    new ValueStateDescriptor[Option[Equipment]]("EquipmentWithModelId", classOf[Option[Equipment]])
  private lazy val equipmentState = getRuntimeContext.getState(equipmentDesc)
  private lazy val linksDesc =
    new ValueStateDescriptor[Set[EquipmentModelLink]]("EquipmentModelLinkAction", classOf[Set[EquipmentModelLink]])
  private lazy val linksState = getRuntimeContext.getState(linksDesc)
  override def processElement1(
      value: EquipmentAction,
      ctx: CoProcessFunction[EquipmentAction, EquipmentModelLinkAction, EquipmentWithModelIdAction]#Context,
      out: Collector[EquipmentWithModelIdAction]): Unit = {
    value match {
      case EquipmentInsert(_, after, _) =>
        equipmentState.update(Some(after))
        Option(linksState.value())
          .map(sets => sets.map(v => EquipmentWithLink(after, v)))
          .map(v => v.map(EquipmentWithModelIdInsert))
          .foreach(v => v.foreach(out.collect))
      case EquipmentUpdate(_, after, _, _) =>
        equipmentState.update(Some(after))
        Option(linksState.value())
          .map(sets => sets.map(v => EquipmentWithLink(after, v)))
          .map(v => v.map(EquipmentWithModelIdInsert))
          .foreach(v => v.foreach(out.collect))
      case EquipmentDelete(_, before, _) =>
        equipmentState.update(None)
        Option(linksState.value())
          .map(sets => sets.map(v => EquipmentWithLink(before, v)))
          .map(v => v.map(EquipmentWithModelIdDelete))
          .foreach(v => v.foreach(out.collect))
    }
  }

  override def processElement2(
      value: EquipmentModelLinkAction,
      ctx: CoProcessFunction[EquipmentAction, EquipmentModelLinkAction, EquipmentWithModelIdAction]#Context,
      out: Collector[EquipmentWithModelIdAction]): Unit = {
    value match {
      case EquipmentModelLinkInsert(after, _) =>
        val links  = Option(linksState.value()).getOrElse(Set())
        val nLinks = links + after
        linksState.update(nLinks)
        val current = Option(equipmentState.value()).flatten
        current.foreach { equipment =>
          val existed = links.filter(_.mid == after.mid)
          if (existed.isEmpty) {
            out.collect(EquipmentWithModelIdInsert(EquipmentWithLink(equipment, after)))
          } else {
            if (!existed.exists(_.dataBinding == after.dataBinding)) {
              out.collect(EquipmentWithModelIdInsert(EquipmentWithLink(equipment, after)))
            }
          }
        }
      case EquipmentModelLinkUpdate(_, after, _) =>
        val links  = Option(linksState.value()).getOrElse(Set())
        val nLinks = links + after
        linksState.update(nLinks)
        val current = Option(equipmentState.value()).flatten
        current.foreach { equipment =>
          val existed = links.filter(_.mid == after.mid)
          if (existed.isEmpty) {
            out.collect(EquipmentWithModelIdInsert(EquipmentWithLink(equipment, after)))
          } else {
            if (!existed.exists(_.dataBinding == after.dataBinding)) {
              out.collect(EquipmentWithModelIdInsert(EquipmentWithLink(equipment, after)))
            }
          }
        }
      case EquipmentModelLinkDelete(before, _) =>
        val links  = Option(linksState.value()).getOrElse(Set())
        val nLinks = links - before
        linksState.update(nLinks)
        val current = Option(equipmentState.value()).flatten
        current.foreach { equipment =>
          if (links.exists(_.mid == before.mid)) {
            out.collect(EquipmentWithModelIdInsert(EquipmentWithLink(equipment, before)))
          }
        }
    }
  }
}
